package com.basics.framework.redission;

import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class RedissonDelayQueueClient implements InitializingBean {


    @Resource
    private RedissonClient redissonClient;

    private final Map<String, RDelayedQueue<DelayMessage>> delayQueueMap = new ConcurrentHashMap<>(16);

    public void addDelayMessage(DelayMessage delayMessage) {
        log.info("delayMessage={}", delayMessage);
        if (delayQueueMap.get(delayMessage.getQueueName()) == null) {
            log.warn("queueName={},该延迟队列不存在，请确认后再试...", delayMessage.getQueueName());
            return;
        }
        RDelayedQueue<DelayMessage> rDelayedQueue = delayQueueMap.get(delayMessage.getQueueName());
        rDelayedQueue.remove(delayMessage);
        rDelayedQueue.offer(delayMessage, delayMessage.getDelayTime(),
                delayMessage.getTimeUnit() == null ? TimeUnit.SECONDS : delayMessage.getTimeUnit());
        redissonClient.getDelayedQueue(rDelayedQueue);
    }


    public void removeDelayMessage(DelayMessage delayMessage) {
        log.info("取消：delayMessage={}", delayMessage);
        if (delayQueueMap.get(delayMessage.getQueueName()) == null) {
            log.warn("queueName={},该延迟队列不存在，请确认后再试...", delayMessage.getQueueName());
            return;
        }

        RDelayedQueue<DelayMessage> rDelayedQueue = delayQueueMap.get(delayMessage.getQueueName());
        rDelayedQueue.remove(delayMessage);
        removeDelayQueue(delayMessage);
    }

    /**
     * 项目重启时操作延时队列，消费因服务停止造成的阻塞的消息
     */
    @PostConstruct
    public void reScheduleDelayedTasks() {
        removeDelayQueue(null);
    }

    /**
     * 删除延迟队列中的元素
     *
     * @param value
     * @param <T>
     * @return
     */
    public void removeDelayQueue(DelayMessage value) {
        RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
        for (RedisDelayQueueEnum queueEnum : queueEnums) {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueEnum.getCode());
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.remove(value);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 有新的延迟队列在这里添加，队列消费类需要继承DelayQueueConsumer
        RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();

        for (RedisDelayQueueEnum queueEnum : queueEnums) {
            DelayQueueConsumer delayQueueConsumer = SpringUtil.getBean(queueEnum.getBeanId());
            if (delayQueueConsumer == null) {
                throw new RuntimeException("queueName=" + queueEnum.getBeanId() + ",delayQueueConsumer=null,请检查配置...");
            }
            // Redisson的延时队列是对另一个队列的再包装，使用时要先将延时消息添加到延时队列中，当延时队列中的消息达到设定的延时时间后，
            // 该延时消息才会进行进入到被包装队列中，因此，我们只需要对被包装队列进行监听即可。
            RBlockingQueue<DelayMessage> rBlockingQueue = redissonClient.getBlockingDeque(queueEnum.getCode());
            RDelayedQueue<DelayMessage> rDelayedQueue = redissonClient.getDelayedQueue(rBlockingQueue);
            delayQueueMap.put(queueEnum.getCode(), rDelayedQueue);
            // 订阅新元素的到来，调用的是takeAsync()，异步执行
            rBlockingQueue.subscribeOnElements(delayQueueConsumer::execute);
        }
    }

}
